Documentación operativa de gss_bi_udfs

Configura una SparkSession global para uso interno del módulo.

from pyspark.sql import SparkSession
from gss_bi_udfs import utils

spark = SparkSession.builder.getOrCreate()
utils.set_spark(spark)

Obtiene Spark en este orden: parámetro explícito, sesión configurada por set_spark, builtins.spark o builder.getOrCreate().

spark = utils.get_spark()

Obtiene el entorno de ejecución y usa un default cuando la variable no está definida.

env = utils.get_env(default="dev")

Si ENV=pro devuelve el catálogo base. En otros casos agrega el sufijo del ambiente.

catalogo = utils.get_env_catalog("fi_comunes")

Concatena el catálogo modificado por ambiente con la ruta schema.tabla.

path = utils.get_env_table_path("fi_comunes", "silver.dim_cliente")

Ejecuta DESCRIBE SCHEMA EXTENDED y extrae el valor RootLocation.

root = utils.get_schema_root_location(
    spark=spark,
    catalog="fi_comunes",
    schema="silver"
)

Devuelve información integral de tabla: nombre completo, path, existencia, provider y tipo.

info = utils.get_table_info(
    spark=spark,
    full_table_name="fi_comunes.silver.dim_cliente"
)
print(info["full_table_name"], info["path"], info["exists"])

Devuelve una Column con valores como -999, N/A, 1900-01-01 o False según el tipo de dato.

default_col = utils.get_default_value_by_type(df.schema["fecha_alta"].dataType)

Está orientada a .csv, .txt y variantes similares leídas de forma nativa con Spark.

from gss_bi_udfs import io

df_clientes = io.load_latest_delimited_file(
    spark=spark,
    source_file="comercial/clientes_vigentes/periodo=2026-03",
    delimiter=";",
    header=True,
    encoding="latin1"
)

Busca el último .xls o .xlsx dentro de la carpeta solicitada.

df_excel = io.load_latest_excel_file(
    spark=spark,
    source_file="comercial/clientes_vigentes/periodo=2026-03",
    header=True,
    sheet_name="Hoja1"
)

Soporta excel, xls, xlsx, csv, txt, tsv y delimited.

df = io.load_latest_file(
    spark=spark,
    source_file="comercial/clientes_vigentes/periodo=2026-03",
    file_format="delimited",
    delimiter="|",
    header=False,
    infer_schema=False
)

Carga el parquet más reciente desde /Volumes/bronze/{db}_{schema}/{env}/{table}/.

df_clientes = io.load_latest_parquet(
    spark=spark,
    data_base="timepro",
    schema="insudb",
    table="clientes",
    env="dev"
)

Permite definir una configuración de tablas y vistas para levantar varias fuentes de una sola vez.

tables_load = {
    "timepro": {
        "insudb": [
            {"table": "clientes", "view": "vw_clientes"}
        ]
    }
}

dfs = io.return_parquets_and_register_temp_views(
    spark=spark,
    tables_load=tables_load,
    verbose=True
)

Misma lógica general de carga que la función anterior, pero sin retornar el diccionario.

io.parquets_register_temp_views(
    spark=spark,
    tables_load=tables_load,
    verbose=True
)

Internamente delega a load_latest_excel_file y retorna None si falla.

df_legacy = io.load_latest_excel(
    spark=spark,
    source_file="comercial/clientes_vigentes/periodo=2026-03",
    env="dev"
)

Es útil cuando un proceso trabaja con varios archivos por dominio y subdominio.

files_load = {
    "comercial": {
        "clientes": [
            {"file": "padron", "view": "vw_padron"}
        ]
    }
}

dfs = io.return_excels_and_register_temp_views(
    spark=spark,
    files_load=files_load,
    verbose=True
)

Es la variante de ejecución más directa cuando solo importa dejar vistas registradas en Spark.

io.excels_register_temp_views(
    spark=spark,
    files_load=files_load,
    verbose=True
)

Permite delegar la ejecución a funciones como return_parquets_and_register_temp_views o excels_register_temp_views.

result = io.load_and_materialize_views(
    "return_parquets_and_register_temp_views",
    spark=spark,
    tables_load=tables_load,
    verbose=True
)

Usa get_table_info para determinar la ubicación y el nombre completo de la tabla destino.

io.save_table_to_delta(
    spark=spark,
    df=df_resultado,
    catalog="fi_comunes",
    schema="silver",
    table_name="dim_cliente",
    mode="overwrite"
)

Está pensada para tablas fact cargadas por período o time_key, validando el rango presente en el DataFrame origen.

io.save_snapshot_to_delta(
    spark=spark,
    df=df_fact,
    catalog="fi_comunes",
    schema="gold",
    table_name="fact_ventas",
    snapshot_column="periodo"
)

Sirve para preparar datasets antes de transformaciones de negocio, normalizando nombres de columnas, limpiando strings y descartando filas vacías.

from gss_bi_udfs.transforms import clean_dataset

df_clean = clean_dataset(
    df=df_raw,
    drop_duplicates=True,
    fillna={"estado": "N/A"}
)

Sirve para crear una PK técnica rápida y determinística a partir de columnas de negocio.

from gss_bi_udfs.transforms import add_hashid

df_hash = add_hashid(
    df=df_clientes,
    columns=["documento", "sucursal"],
    new_col_name="cliente_hash"
)

Usa get_default_value_by_type para asignar un valor técnico por cada tipo del esquema.

from gss_bi_udfs.transforms import get_default_record

df_default = get_default_record(spark, df_dim_cliente)